package com.flink.process.function;

import com.flink.datasource.UserSource;
import com.flink.entity.User;
import com.flink.window.WindowFunctionDemo;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;

import java.sql.Timestamp;
import java.time.Duration;

/**
 * 描述:
 * 事件时间 处理函数 定时器
 *
 * @author yanzhengwu
 * @create 2022-08-07 17:18
 */
public class EventTimeTimerFunctionDemo {

    public static void main(String[] args) throws Exception {

        //声明执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //测试为了保证元素顺序并行度设置为1，可以理解为单线程执行
        env.setParallelism(1);
        //设置水位线生成的间隔 这里给的是100毫秒 ,flink 默认是200毫秒 ，flink 可以达到毫秒级别的效率
//        env.getConfig().setAutoWatermarkInterval(100);


        //TODO 无序流的watermark生成策略
        DataStream<User> stream = env.addSource(new UserSource())       //生成水位线和时间戳的策略对象
                .assignTimestampsAndWatermarks(
                        //返回一个具体的策略对象(TODO 这里是乱序流的处理方法，给了一个延迟2秒的策略，也可由理解为 数据延迟多长时间能够全部到位)
                        WatermarkStrategy.<User>forBoundedOutOfOrderness(Duration.ofSeconds(2L))
                                //返回策略对象的具体实现
                                .withTimestampAssigner(new SerializableTimestampAssigner<User>() {
                                    /**
                                     * 此方法为指定以事件时间的具体时间戳字段
                                     *
                                     * @param element
                                     * @param recordTimestamp
                                     * @return 返回的则是一个毫秒数的时间戳
                                     */
                                    @Override
                                    public long extractTimestamp(User element, long recordTimestamp) {
                                        return element.getTimestamp();
                                    }
                                }));
        stream.keyBy(User::getName)
                .process(new KeyedProcessFunction<String, User, String>() {
                    @Override
                    public void processElement(User value, Context ctx, Collector<String> out) throws Exception {
                        //TODO 注意：这里不要使用水位线时间戳，水位线生成是有延迟的 ，这里直接使用数据时间
                        long l = value.getTimestamp();

                        out.collect(ctx.getCurrentKey()+"->数据到达时间：" + new Timestamp(l));
                        //TODO 注册一个定时器 (根据分组的key 10秒统计一次,将当前处理时间 +上10秒，注意：不是事件时间而是系统处理时间)
                        ctx.timerService().registerEventTimeTimer(l + 10 * 1000L);
                    }

                    @Override
                    public void onTimer(long timestamp, OnTimerContext ctx, Collector<String> out) throws Exception {
                        out.collect(ctx.getCurrentKey()+"->定时器触发时间："+new Timestamp(timestamp));
                    }


                }).print();
        //TODO 注意时间是根据水位线进行触发定时器 ，所以数据会出现延迟 属于正常现象，水位线由于滞后2秒 所以会触发晚两秒
        //TODO 如果所有数据根据watermark在定时器触发之前都已处理完毕或已到达，flink引擎会将watermark推进到long.max将所有定时器全部触发！

        env.execute();
    }
}
